package com.lagou.hbase;

import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

public class HbaseApiDemo {

    public static final String TABLE_NAME = "user_friends";
    public static final String FAMILY = "friends";

    public static void main(String[] args) throws Exception {

        /*
         * 获取hbase连接
         */
        Connection connection = new HbaseClient().getConnection();
        createTable(connection);
        putData(connection);

        // 挂载协处理器
        LoadProcessor loadProcessor = new LoadProcessor(connection, TableName.valueOf(TABLE_NAME));
        loadProcessor.loadProcess();

        // 删除测试
        deleteData(connection);

        connection.close();
    }

    /**
     * 创建表 user_friendss
     * 并创建列族 friends
     *
     * @param connection
     * @throws IOException
     */
    private static void createTable(Connection connection) throws IOException {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();

        // 删除掉已经存在的表, 方便测试
        boolean tableAvailable = admin.isTableAvailable(tableName);
        if (tableAvailable) {
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }
        HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
        hTableDescriptor.addFamily(new HColumnDescriptor(FAMILY));
        admin.createTable(hTableDescriptor);
        admin.close();
        System.out.println("创建" + TABLE_NAME + "成功  ");
    }

    /**
     * 向表中批量插入测试数据
     *
     * @param connection
     */
    private static void putData(Connection connection) throws IOException {
        List<Put> puts = new ArrayList<>();
        Random random = new Random();
        Table userFriends = connection.getTable(TableName.valueOf(TABLE_NAME));
        for (int i = 1; i <= 10; i++) {
            // rowKey 为 uid开头
            Put put = new Put(Bytes.toBytes("uid_" + i));

            // 随机生成不同数量的好友绑定关系
            for (int j = 1; j < random.nextInt(5) + 2; j++) {
                // 随机生成好友关系
                String friends = "uid_" + (random.nextInt(10) + 1);
                put.addColumn(Bytes.toBytes(FAMILY), Bytes.toBytes(friends), Bytes.toBytes(friends));

                // 双向绑定关系
                Put friendsPut = new Put(Bytes.toBytes(friends));
                friendsPut.addColumn(Bytes.toBytes(FAMILY), Bytes.toBytes("uid_" + i), Bytes.toBytes("uid_" + i));
                puts.add(friendsPut);
            }
            puts.add(put);

        }


        System.out.println("## 表结构数据展示  \n  ");
        userFriends.put(puts);
        Scan scan = new Scan();
        ResultScanner scanner = userFriends.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                String family = Bytes.toString(CellUtil.cloneFamily(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.printf("rowKey: %s, family:%s, value:%s%n", rowKey, family, value);
            }
        }
        userFriends.close();
    }

    /**
     * 删除数据测试
     * 随机取一条数据进行删除测试
     * 从自己的数据中删除朋友的数据
     * 然后从俩个数据中获取数据测试
     */
    private static void deleteData(Connection connection) throws Exception {
        Table table = connection.getTable(TableName.valueOf(TABLE_NAME));
        int randUid = new Random().nextInt(10) + 1;
        String uid = "uid_" + randUid;
        Result result = table.get(new Get(Bytes.toBytes(uid)));
        String friendsUid = null;
        if (result.rawCells().length > 0) {
            friendsUid = Bytes.toString(CellUtil.cloneValue(result.rawCells()[0]));
            System.out.println("获取到好友数据->" + friendsUid);
        }
        if (null == friendsUid) {
            table.close();
            return;
        }
        System.out.printf("准备删除好友-> uid:%s friend:%s %n  ", uid, friendsUid);
        // 删除前数据严重
        deleteDiff(uid, friendsUid, table);


        Delete delete = new Delete(Bytes.toBytes(uid));
        delete.addColumn(Bytes.toBytes(FAMILY), Bytes.toBytes(friendsUid));
        table.delete(delete);
        System.out.println("删除完成... " + delete.toString());

        // 删除后验证
        deleteDiff(uid, friendsUid, table);
        table.close();
    }

    private static void deleteDiff(String uid, String friendsUid, Table table) throws IOException {
        // 删除前数据获取验证
        Get get = new Get(Bytes.toBytes(uid));
        get.addColumn(Bytes.toBytes(FAMILY), Bytes.toBytes(friendsUid));
        Result result1 = table.get(get);

        System.out.printf("删除前数据验证 uid:%s -> friend:%s exists:%s,length:%s %s %n   ", uid, friendsUid, Arrays.stream(result1.rawCells()).map(s -> CellUtil.toString(s, true)).collect(Collectors.toList()), result1.rawCells().length, get.toString());

        get = new Get(Bytes.toBytes(friendsUid));
        get.addColumn(Bytes.toBytes(FAMILY), Bytes.toBytes(uid));
        Result result2 = table.get(get);
        System.out.printf("删除前数据验证 uid:%s -> friend:%s exists:%s,length:%s %s %n  ", friendsUid, uid, Arrays.stream(result2.rawCells()).map(s -> CellUtil.toString(s, true)).collect(Collectors.toList()), result1.rawCells().length, get.toString());

    }

}
